package org.common.utils.consumer;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * 消费组
 * @author zhouzhibing
 * @date 2025/8/1
 */
public class ConsumerGroup {

    /**
     * 队列列表
     */
    private List<ConsumerQueue> queueList;

    private boolean start;

    /**
     * 消费对象组
     * @param config 配置
     */
    ConsumerGroup(ConsumerConfig config) {
        Objects.requireNonNull(config);
        Objects.requireNonNull(config.getExecutor());
        Objects.requireNonNull(config.getHandler());

        int count = config.getCount();
        queueList = new ArrayList<>(count);
        for (int i = 0; i < count; i++) {
            ConsumerQueue consumerQueue = new ConsumerQueue(config , i);
            consumerQueue.start();
            queueList.add(consumerQueue);
        }
        start = true;
    }

    /**
     * 添加异步对象
     * @param entity 异步对象
     */
    public void add(Object entity) {
        if(!start) {
            throw new ConsumerException("ConsumerGroup.add is close , entity = " + entity);
        }

        if(queueList.size() > 1) {
            int index = Math.abs(Objects.hash(entity) % queueList.size());
            ConsumerQueue consumerQueue = queueList.get(index);
            consumerQueue.add(entity);
        } else {
            ConsumerQueue consumerQueue = queueList.getFirst();
            consumerQueue.add(entity);
        }
    }

    /**
     * 关闭
     */
    public void close() {
        start = false;
        for (ConsumerQueue consumerQueue : queueList) {
            consumerQueue.close();
        }
        queueList.clear();
    }
}
